package com.express.gateway.verticle;

import com.express.gateway.common.CommonCons;
import com.express.gateway.common.EventBusTopicConst;
import com.express.gateway.common.ResultFormat;
import com.express.gateway.manage.entity.DeployInfo;
import com.express.gateway.manage.enums.StatusCodeEnum;
import com.express.gateway.manage.enums.TransProtolEnum;
import com.express.gateway.manage.options.ServerOptions;
import io.vertx.core.*;
import io.vertx.core.eventbus.Message;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.net.BindException;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.stream.Stream;

import static com.express.gateway.common.CommonCons.KEY_CLUSTER;

/**
 * 用于部署应用与部署API
 *
 * @author flower
 */
public class AppCtrlVerticle extends AbstractVerticle {
    private static final Logger LOG = LogManager.getLogger(AppCtrlVerticle.class);
    /**
     * 存储已经在运行了的项目
     */
    private final Map<String, DeployInfo> runAppMaps = new HashMap<>();
    /**
     * 存储已经在运行的应用API集合
     */
    private final Map<String, Set<String>> runApiMaps = new HashMap<>();
    /**
     * 端口服务代理者
     */
    private final Map<Integer, DeployInfo> portProxyMap = new HashMap<>();
    /**
     * 备用端口服务代理者
     */
    private final Map<Integer, List<DeployInfo>> portStandbyProxyMap = new HashMap<>();
    /**
     * 被端口服务代理的应用,set值为端口号+appName
     */
    private Map<Integer, Set<String>> portProxyAppMap = new HashMap<>();
    /**
     * 被端口服务代理应用的API,key值为端口号+appName
     */
    private Map<String, JsonArray> portProxyApiMap = new HashMap<>();

    @Override
    public void start(Promise<Void> startFuture) throws Exception {
        LOG.info("start Deployment Verticle");
        // 如果vert.x是以集群的方式运行添加接受广播的部署相关
        if (vertx.isClustered()) {
            vertx.eventBus().consumer(KEY_CLUSTER + EventBusTopicConst.DEPLOY_APP_DEPLOY, this::deploymentApp);
            vertx.eventBus().consumer(KEY_CLUSTER + EventBusTopicConst.DEPLOY_APP_UNDEPLOY, this::unDeploymentApp);
            vertx.eventBus().consumer(KEY_CLUSTER + EventBusTopicConst.DEPLOY_API_START_ALL, this::startAllApi);
            vertx.eventBus().consumer(KEY_CLUSTER + EventBusTopicConst.DEPLOY_API_START, this::startApi);
            vertx.eventBus().consumer(KEY_CLUSTER + EventBusTopicConst.DEPLOY_API_STOP, this::stopApi);
        }

        vertx.eventBus().consumer(EventBusTopicConst.DEPLOY_APP_COUNT, this::applicationCount);

        vertx.eventBus().consumer(EventBusTopicConst.DEPLOY_APP_DEPLOY, this::deploymentApp);
        vertx.eventBus().consumer(EventBusTopicConst.DEPLOY_APP_UNDEPLOY, this::unDeploymentApp);
        vertx.eventBus().consumer(EventBusTopicConst.DEPLOY_API_START_ALL, this::startAllApi);
        vertx.eventBus().consumer(EventBusTopicConst.DEPLOY_API_START, this::startApi);
        vertx.eventBus().consumer(EventBusTopicConst.DEPLOY_API_STOP, this::stopApi);

        vertx.eventBus().consumer(EventBusTopicConst.DEPLOY_FIND_ONLINE_APP, this::findOnlineApp);
        vertx.eventBus().consumer(EventBusTopicConst.DEPLOY_APP_IS_ONLINE, this::getAppIsOnline);
        vertx.eventBus().consumer(EventBusTopicConst.DEPLOY_FIND_ONLINE_API, this::findOnlineApi);
        vertx.eventBus().consumer(EventBusTopicConst.DEPLOY_API_IS_ONLINE, this::getApiIsOnline);
        LOG.info("start Deployment Verticle successful");
        super.start(startFuture);
    }

    /**
     * 部署应用程序
     *
     * @param msg
     */
    public void deploymentApp(Message<JsonObject> msg) {
        JsonObject body = new JsonObject();
        if (vertx.isClustered() && isSamApp(msg.body())) {
            return;
        }
        String name = msg.body().getString(CommonCons.KEY_APP_NAME);
        JsonObject application = msg.body().getJsonObject(CommonCons.KEY_APP);
        body.put("appConfig", application);

        // S001-部署应用服务器, 启动后部署数据库查询出应用配置服务
        vertx.eventBus().<JsonArray>request(EventBusTopicConst.SYSTEM_BLACK_IP_FIND, null, list -> {
            if (list.failed()) {
                msg.fail(500, list.cause().toString());
                return;
            }
            // 添加到加载配置
            body.put("blackIpSet", list.result().body());
            DeploymentOptions options = new DeploymentOptions(config());
            // options.setIsolationGroup(name);
            options.setConfig(body);
            deployApps(msg, options, name, application);
        });

    }

    private void deployApps(Message<JsonObject> msg, DeploymentOptions options, String name, JsonObject application) {
        vertx.deployVerticle(AppExecuteVerticle.class.getName(), options).onFailure(ex -> {
            LOG.info("启动应用:{} 结果:{}", name, false, ex.getCause());
            StatusCodeEnum status = StatusCodeEnum.C500;
            if (ex.getCause() instanceof BindException) {
                status = StatusCodeEnum.C1111;
            }
            msg.fail(status.getCode(), ex.getMessage());
        }).onSuccess( rs -> {
            ServerOptions serverOptions = ServerOptions.fromJson(application.getJsonObject("serverOptions"));
            DeployInfo deployInfo = new DeployInfo(name, rs, serverOptions);
            runAppMaps.put(name, deployInfo);
            runApiMaps.put(name, new HashSet<>());
            // 设置端口服务号代理
            // http端口号
            Integer httpPort = deployInfo.getHttpPort();
            recordDeployServerInfo(httpPort, deployInfo);
            // https端口号
            Integer httpsPort = deployInfo.getHttpsPort();
            recordDeployServerInfo(httpsPort, deployInfo);
            msg.reply("ok");
        });
    }

    private void recordDeployServerInfo(Integer servPort, DeployInfo deploy) {
        Optional.ofNullable(servPort).ifPresent(port -> {
            if (portProxyMap.get(port) == null) {
                portProxyMap.put(port, deploy);
                return;
            }
//            List<DeployInfo> item = Optional.ofNullable(portStandbyProxyMap.get(port)).orElse(new ArrayList<>());
//            item.add(deploy);
//            portStandbyProxyMap.put(port, item);
            portStandbyProxyMap.computeIfAbsent(port, k -> new ArrayList<>()).add(deploy);
        });
    }

    /**
     * 卸载应用程序
     *
     * @param msg
     */
    public void unDeploymentApp(Message<JsonObject> msg) {
        if (vertx.isClustered() && isSamApp(msg.body())) {
            return;
        }
        String name = msg.body().getString(CommonCons.KEY_APP_NAME);
        DeployInfo deployInfo = runAppMaps.get(name);
        if (deployInfo == null) {
            msg.reply("ok");
            return;
        }
        String deployId = deployInfo.getDeployId();

        Promise<Void> unDeployFuture = Promise.promise();
        unDeployFuture.future().onComplete(handle -> {
            vertx.undeploy(deployId).onFailure(ex -> {
                LOG.error("暂停应用:" + name + " 失败:" + ex.getCause());
                msg.fail(500, ex.getMessage());
            }).onSuccess(rs -> {
                vertx.executeBlocking(future -> {
                    LOG.info("暂停应用:" + name + " 成功!");
                    // 设置备用端口服务代理为主端口服务代理
                    stopNetServEndpoint(deployInfo);
                    // stopHttpsEndpoint(deployInfos);
                    future.complete();
                }, futures -> {
                    runAppMaps.remove(deployInfo.getAppName());
                    runApiMaps.remove(deployInfo.getAppName());
                    msg.reply("ok");
                });
            });
        });
        // 暂停API后暂停应用
        vertx.executeBlocking(promise -> {
            stopApiRecursion(new ArrayList<>(runApiMaps.get(deployInfo.getAppName())), deployInfo.getAppName(), 0, 0, res -> {
                JsonObject result = res.result() == null ? new JsonObject() : res.result();
                int success = result.getInteger(CommonCons.KEY_SUCCESS, 0);
                int fail = result.getInteger(CommonCons.KEY_FAIL, 0);
                LOG.info("暂停应用：{} ,所有 API 结果:成功 {} 条, 失败 {} 条", deployInfo.getAppName(), success, fail);
                promise.complete();
            });
        }, unDeployFuture);
    }


    private void stopNetServEndpoint(DeployInfo deployInfo) {
        boolean isHttps = TransProtolEnum.HTTPS.equals(deployInfo.getTransProto());
        Integer httpPort = isHttps ? deployInfo.getHttpsPort() : deployInfo.getHttpPort();
        DeployInfo proxy = portProxyMap.get(httpPort);
        if (Stream.of(httpPort, proxy).anyMatch(Objects::isNull)) {
            return;
        }
        Integer proxyPort = isHttps ? proxy.getHttpsPort() : proxy.getHttpPort();
        if (proxyPort == null) {
            return;
        }

        portProxyAppMap = Optional.ofNullable(portProxyAppMap).orElse(new HashMap<>());
        portProxyApiMap = Optional.ofNullable(portProxyApiMap).orElse(new HashMap<>());

        // 查询代理类中是否有启动当前应用的API
        Set<String> proxySet = portProxyAppMap.get(proxyPort);
        String proxyKey = httpPort + deployInfo.getAppName();
        if (proxySet != null && !deployInfo.getAppName().equals(proxy.getAppName()) && proxySet.contains(proxyKey)) {
            proxySet.remove(proxyKey);
            portProxyAppMap.put(proxyPort, proxySet);
            portProxyApiMap.remove(proxyKey);
            Set<String> apiNames = runApiMaps.get(deployInfo.getAppName());
            stopApiRecursion(new ArrayList<>(apiNames), proxy.getAppName(), 0, 0, stopRes -> {
                JsonObject result = stopRes.result() == null ? new JsonObject() : stopRes.result();
                int success = result.getInteger(CommonCons.KEY_SUCCESS, 0);
                int fail = result.getInteger(CommonCons.KEY_FAIL, 0);
                LOG.info("应用:{} 协议:{} 暂停:{}所有API 操作API, 成功:{},失败:{}", proxy.getAppName(), deployInfo.getTransProto(), deployInfo.getAppName(), success, fail);
            });
        }
        // 设置备用代理为主代理并将代理的API移动转交给备用代理类,如果没有备用类则当前代理
        if (!deployInfo.getAppName().equals(proxy.getAppName())) {
            return;
        }
        List<DeployInfo> item = portStandbyProxyMap.get(proxyPort);
        if (item == null || item.isEmpty()) {
            portProxyMap.remove(proxyPort);
            return;
        }
        DeployInfo standby = item.remove(0);
        Set<String> set = portProxyAppMap.get(proxyPort);
        if (set == null) {
            portProxyMap.put(proxyPort, standby);
            return;
        }
        String standbyAppName = standby.getAppName();
        JsonArray body = new JsonArray();
        set.forEach(k -> {
            body.addAll(portProxyApiMap.get(k));
        });
        startAllApiRecursion(body, standbyAppName, 0, null, 0, true, 1, startProxy -> {
            JsonObject result = startProxy.result() == null ? new JsonObject() : startProxy.result();
            int success = result.getInteger(CommonCons.KEY_SUCCESS, 0);
            int fail = result.getInteger(CommonCons.KEY_FAIL, 0);
            LOG.info("代理:{} 协议:{} API迁移到:{} 结果, 成功:{},失败:{}", proxy.getAppName(), deployInfo.getTransProto(), standbyAppName, success, fail);
        });
    }

    /**
     * S002-启动所有API
     *
     * @param msg
     */
    public void startAllApi(Message<JsonObject> msg) {
        if (vertx.isClustered() && isSamApp(msg.body())) {
            return;
        }
        String appName = msg.body().getString(CommonCons.KEY_APP_NAME);
        DeployInfo deployInfo = runAppMaps.get(appName);
        if (deployInfo == null) {
            msg.reply(ResultFormat.format(StatusCodeEnum.C1400, "应用尚未启动"));
            LOG.info("启动:{} 内所有API失败: 应用未启动", appName );
            return;
        }
        JsonArray body = msg.body().getJsonArray("apis");
        if (body == null || !body.isEmpty()) {
            msg.reply(ResultFormat.format(StatusCodeEnum.C200, "API数量为: 0"));
            LOG.debug("启动:{}, 内所有API结果: API数量为 0", appName);
            return;
        }

        runApiMaps.computeIfAbsent(appName, k -> new HashSet<>());
        // 启动所有API
        vertx.<String>executeBlocking(fut -> {
            // 代理是否启动http服务
            // 代理是否启动https服务
            startInterface(deployInfo, body, appName);
            // 启动成功的API集合
            Set<String> successSet = new HashSet<>();
            startAllApiRecursion(body.copy(), appName, 0, successSet, 0, res -> {
                JsonObject result = res.result() == null ? new JsonObject() : res.result();
                int success = result.getInteger(CommonCons.KEY_SUCCESS, 0);
                int fail = result.getInteger(CommonCons.KEY_FAIL, 0);
                runApiMaps.put(appName, successSet);
                LOG.info("启动:{} 所有API结果: 成功:{} ,失败:{}", appName, success, fail);
                fut.complete(ResultFormat.format(StatusCodeEnum.C200, "启动" + appName + "所有API结果: 成功数量:" + success + ",失败数量:" + fail));
            });
        }, res -> {
            msg.reply(res.result());
        });


    }

    /**
     * 启动所有API的方法
     *
     * @param apis       apis集合
     * @param appName    应用的名称
     * @param success    启动成功的种子
     * @param successSet 启动成功的API,如果successSet等于null,则不添加成功的API
     * @param fail       启动失败的种子
     * @return 返回json key : success成功的数量,fail失败的数量
     */
    private void startAllApiRecursion(JsonArray apis, String appName, int success, Set<String> successSet, int fail, Handler<AsyncResult<JsonObject>> handler) {
        startAllApiRecursion(apis, appName, success, successSet, fail, null, null, handler);
    }

    /**
     * 启动所有API的方法
     *
     * @param apis            apis集合
     * @param appName         应用的名称
     * @param success         启动成功的种子
     * @param successSet      启动成功的API,如果successSet等于null,则不添加成功的API
     * @param fail            启动失败的种子
     * @param elseRouteToThis 是否代理应用启动
     * @param serverType      服务的类型 1=http, 2=https, 3=webSocket
     * @param handler         回调函数
     */
    private void startAllApiRecursion(JsonArray apis, String appName, int success, Set<String> successSet, int fail, Boolean elseRouteToThis, Integer serverType, Handler<AsyncResult<JsonObject>> handler) {
        // 全部启动完后 apis 会被清空
        if (Objects.isNull(apis) || apis.isEmpty()) {
            JsonObject result = new JsonObject();
            result.put(CommonCons.KEY_SUCCESS, success);
            result.put(CommonCons.KEY_FAIL, fail);
            handler.handle(Future.succeededFuture(result));
            return;
        }
        JsonObject api = (JsonObject) apis.remove(0);
        JsonObject params = new JsonObject().put(CommonCons.KEY_API_NAME, api);
        Optional.ofNullable(serverType).ifPresent(d -> params.put("serverType", d));
        Optional.ofNullable(elseRouteToThis).ifPresent(d -> params.put("elseRouteToThis", d));
        // SOO3-调用AppVerticle addRoute 启动路由
        String adds = appName + EventBusTopicConst.APPLICATION_ADD_API_SUFFIX;
        vertx.eventBus().request(adds, params, reply -> {
            if (reply.succeeded()) {
                Optional.ofNullable(successSet).ifPresent(set -> set.add(api.getString(CommonCons.KEY_API_NAME)));
                startAllApiRecursion(apis, appName, success + 1, successSet, fail, elseRouteToThis, serverType, handler);
            } else {
                LOG.error("启动：{} 所有API, 失败:{}" ,api.getString(CommonCons.KEY_API_NAME) ,reply.cause());
                startAllApiRecursion(apis, appName, success, successSet, fail + 1, elseRouteToThis, serverType, handler);
            }
        });

    }

    /**
     * 启动API
     *
     * @param msg
     */
    public void startApi(Message<JsonObject> msg) {
        if (vertx.isClustered() && isSamApp(msg.body())) {
            return;
        }
        String appName = msg.body().getString(CommonCons.KEY_APP_NAME);
        String apiName = msg.body().getString(CommonCons.KEY_API_NAME);
        DeployInfo deployInfo = runAppMaps.get(appName);
        if (deployInfo == null) {
            msg.reply(-1);
            return;
        }
        JsonObject body = msg.body().getJsonObject(CommonCons.KEY_API);
        vertx.<Integer>executeBlocking(future -> {
            JsonArray apis = new JsonArray(Collections.singletonList(body));
            startInterface(deployInfo, apis, appName);
            startApiService(body.copy(), appName, null, deployInfo.getTransProto().getCode(), res -> {
                if (res.failed()) {
                    LOG.error("启动:{}, API:{}, 失败:", appName, apiName, res.cause());
                    future.fail(res.cause());
                    return;
                }
                LOG.error("启动:{}, API:{}, 成功!", appName, apiName);
                future.complete(res.result().body());
            }, null);
        }, res -> {
            if (res.failed()) {
                LOG.error("启动 APP:{}, API:{}, 失败:", appName, apiName, res.cause());
                msg.fail(500, res.cause().getMessage());
                return;
            }
            LOG.error("启动 APP:{}, API:{}, 成功!", appName, apiName);
//            Set<String> set = runApiMaps.get(appName) == null ? new HashSet<>() : runApiMaps.get(appName);
//            set.add(apiName);
//            runApiMaps.put(appName, set);
            runApiMaps.computeIfAbsent(appName, k -> new HashSet<>()).add(apiName);
            msg.reply(1);
        });
    }


    private void startInterface(DeployInfo deployInfo, JsonArray body, String appName) {
        boolean isHttps = TransProtolEnum.HTTPS.equals(deployInfo.getTransProto());
        Integer httpPort = isHttps ? deployInfo.getHttpsPort() : deployInfo.getHttpPort();
        DeployInfo proxy = portProxyMap.get(httpPort);
        if (Stream.of(httpPort, proxy).anyMatch(Objects::isNull)) {
            return;
        }
        Integer proxyPort = isHttps ? proxy.getHttpsPort() : proxy.getHttpPort();
        if (!httpPort.equals(proxyPort) || proxy.getAppName().equals(deployInfo.getAppName())) {
            return;
        }
        startAllApiRecursion(body.copy(), proxy.getAppName(), 0, null, 0, true, 1, res -> {
            JsonObject result = res.result() == null ? new JsonObject() : res.result();
            int success = result.getInteger(CommonCons.KEY_SUCCESS, 0);
            int fail = result.getInteger(CommonCons.KEY_FAIL, 0);
            LOG.info("应用:{} 协议:{} 启动:{} 所有API结果: 成功:{},失败:{}", proxy.getAppName(), deployInfo.getTransProto(), appName, success, fail);
//            Set<String> item = portProxyAppMap.get(proxyPort) == null ? new HashSet<>() : portProxyAppMap.get(proxyPort);
//            item.add(proxyPort + appName);
//            portProxyAppMap.put(proxyPort, item);
            portProxyAppMap.computeIfAbsent(proxyPort, k -> new HashSet<>()).add(proxyPort + appName);
        });
    }

    /**
     * 启动API的服务,该服务为递归会阻塞线程,请在executeBlocking中执行,startResult参数传入null
     *
     * @param body            引用信息
     * @param appName         应用的名称
     * @param elseRouteToThis 是否代理启动
     * @param serverType      服务类型1=HTTP服务,2=HTTPS服务,3=webSocket服务
     * @param handler         返回结果
     * @param startResult     启动的结果,调用时需要传入null
     */
    private void startApiService(JsonObject body, String appName, Boolean elseRouteToThis, Integer serverType, Handler<AsyncResult<Message<Integer>>> handler, AsyncResult<Message<Integer>> startResult) {
        if (startResult != null) {
            handler.handle(startResult);
            return;
        }
        JsonObject params = new JsonObject();
        params.put(CommonCons.KEY_API, body);
        Optional.ofNullable(serverType).ifPresent(d -> params.put("serverType", d));
        Optional.ofNullable(elseRouteToThis).ifPresent(d -> params.put("elseRouteToThis", d));
        String address = appName + EventBusTopicConst.APPLICATION_ADD_API_SUFFIX;
        vertx.eventBus().<Integer>request(address, params, res -> {
            startApiService(body, appName, elseRouteToThis, serverType, handler, res);
        });
    }

    /**
     * 停止一个API
     *
     * @param msg
     */
    private void stopApi(Message<JsonObject> msg) {
        if (vertx.isClustered() && isSamApp(msg.body())) {
            return;
        }
        String appName = msg.body().getString(CommonCons.KEY_APP_NAME);
        String apiName = msg.body().getString(CommonCons.KEY_API_NAME);
        DeployInfo deployInfo = runAppMaps.get(appName);
        if (deployInfo == null) {
            msg.reply(1);
            return;
        }
        vertx.<Integer>executeBlocking(promise -> {
            stopInterface(promise, deployInfo, apiName);
            stopApiServiceSingle(appName, apiName, res -> {
                if (res.succeeded()) {
                    promise.fail(res.cause());
                    return;
                }
                LOG.info(appName + "暂停API: {}, 成功" ,apiName);
                promise.complete(res.result().body());
            }, null);
        }, res -> {
            if (res.succeeded()) {
                LOG.error(appName + "暂停API: {}, 失败" ,apiName, res.cause());
                msg.fail(500, res.cause().toString());
                return;
            }
            if (runApiMaps.get(appName) != null) {
                runApiMaps.get(appName).remove(apiName);
            }
            LOG.info(appName + "暂停API: {} 成功" ,apiName);
            msg.reply(1);
        });
    }

    private void stopInterface(Promise<Integer> promise, DeployInfo deployInfo, String apiName) {
        boolean isHttpsPort = TransProtolEnum.HTTP.equals(deployInfo.getTransProto());
        Integer httpPort = isHttpsPort ? deployInfo.getHttpsPort() : deployInfo.getHttpPort();
        if (httpPort == null) {
            return;
        }
        DeployInfo proxy = portProxyMap.get(httpPort);
        if (Objects.isNull(proxy)) {
            return;
        }
        Integer proxyPort = isHttpsPort ? proxy.getHttpsPort() : proxy.getHttpPort();
        if (!httpPort.equals(proxyPort) || deployInfo.getAppName().equals(proxy.getAppName())) {
            return;
        }
        stopApiServiceSingle(proxy.getAppName(), apiName, res -> {
            if (res.succeeded()) {
                LOG.info("应用:{}, 协议：{} 暂停API:{} 成功", proxy.getAppName(), proxy.getTransProto(), apiName);
                return;
            }
            LOG.error("应用:{}, 协议：{} 暂停API:{} 失败", proxy.getAppName(), proxy.getTransProto(), apiName, res.cause());
            promise.fail(res.cause());
        }, null);
    }

    /**
     * 停止一个API的服务,该服务为递归会阻塞线程,请在executeBlocking中执行,stopResult参数传入null
     *
     * @param appName    应用的名称
     * @param apiName    API的名称
     * @param handler    操作结果
     * @param stopResult 停止的结果,调用时需要传入null
     */
    public void stopApiServiceSingle(String appName, String apiName, Handler<AsyncResult<Message<Integer>>> handler, AsyncResult<Message<Integer>> stopResult) {
        if (stopResult != null) {
            handler.handle(stopResult);
            return;
        }
        vertx.eventBus().<Integer>request(appName + EventBusTopicConst.APPLICATION_DEL_API_SUFFIX, apiName, reply -> {
            stopApiServiceSingle(null, null, handler, reply);
        });
    }

    /**
     * 批量暂停API
     *
     * @param apis    API的名字集
     * @param appName 应用的名称
     * @param success 暂停成功的数量
     * @param fail    暂停失败的数量
     * @param handler
     */
    public void stopApiRecursion(List<String> apis, String appName, int success, int fail, Handler<AsyncResult<JsonObject>> handler) {
        if (apis == null || apis.isEmpty()) {
            JsonObject result = new JsonObject();
            result.put(CommonCons.KEY_SUCCESS, success);
            result.put(CommonCons.KEY_FAIL, fail);
            handler.handle(Future.succeededFuture(result));
            return;
        }
        String apiName = apis.remove(0);
        String address = appName + EventBusTopicConst.APPLICATION_DEL_API_SUFFIX;
        vertx.eventBus().request(address, apiName, reply -> {
            if (reply.succeeded()) {
                stopApiRecursion(apis, appName, success + 1, fail, handler);
            } else {
                stopApiRecursion(apis, appName, success, fail + 1, handler);
            }
        });
    }

    /**
     * 查看所有在线的APP
     *
     * @param msg
     */
    public void findOnlineApp(Message<JsonObject> msg) {
        Set<String> set = Optional.of(runApiMaps.keySet()).orElse(new HashSet<>());
        msg.reply(new JsonArray(new ArrayList<>(set)));
    }

    /**
     * 查看APP是否在线
     *
     * @param msg
     */
    public void getAppIsOnline(Message<String> msg) {
        msg.reply(runAppMaps.get(msg.body()) != null);
    }

    /**
     * 查看所有在线的API
     */
    public void findOnlineApi(Message<JsonObject> msg) {
        String appName = msg.body().getString(CommonCons.KEY_APP_NAME);
        Set<String> set = runApiMaps.get(appName);
        if (set == null) {
            msg.reply(new JsonArray());
        } else {
            msg.reply(new JsonArray(new ArrayList<>(set)));
        }
    }

    /**
     * 查看API是否在线
     *
     * @param msg
     */
    public void getApiIsOnline(Message<JsonObject> msg) {
        String appName = msg.body().getString(CommonCons.KEY_APP_NAME);
        String apiName = msg.body().getString(CommonCons.KEY_API_NAME);
        if (runApiMaps.get(appName) == null) {
            msg.reply(false);
        } else {
            msg.reply(runApiMaps.get(appName).contains(apiName));
        }
    }

    /**
     * 查看在线APP与API的数量
     *
     * @param msg
     */
    public void applicationCount(Message<JsonObject> msg) {
        JsonObject result = new JsonObject();
        result.put(CommonCons.KEY_APP, runAppMaps.size());
        int sum = runApiMaps.values().stream().mapToInt(Set::size).sum();
        result.put(CommonCons.KEY_API, sum);
        msg.reply(result);
    }

    private boolean isSamApp(JsonObject body) {
        String sysId = System.getProperty(CommonCons.KEY_VERTX, CommonCons.KEY_VERTX_NAME);
        return sysId.equals(body.getString(CommonCons.KEY_VERTX));
    }

}
